package com.dtruth.hadoop.hdfs2hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class Hdfs2HBase {

	public static class MapperClass extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
		// 列名
		public static final String[] COLUMNS = { "", "ip", "count" };

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] cloumnVals = value.toString().split(",");
			String rowkey = cloumnVals[1];
			Put put = new Put(rowkey.getBytes());
			for (int i = 1; i < cloumnVals.length; i++) {
				put.add("info".getBytes(), COLUMNS[i].getBytes(), cloumnVals[i].getBytes());
			}
			context.write(new ImmutableBytesWritable(rowkey.getBytes()), put);
		}
	}
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", "hdfs://land:9000/");
		// conf.set("mapreduce.framework.name", "yarn");
		conf.set("mapred.job.tracker", "land:9001");
		conf.set("hbase.zookeeper.quorum", "math");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.master", "math:60000");
		Job job = Job.getInstance(conf, "file2hbase");
		job.setJarByClass(Hdfs2HBase.class);
		job.setMapperClass(MapperClass.class);
		job.setNumReduceTasks(0);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "mytest");
		FileInputFormat.addInputPath(job, new Path("hdfs://land:9000/mrproject/test/out/list"));
		System.out.println(job.waitForCompletion(true) ? 0 : 1);
	}
}